package com.atguigu.flink.datastreamapi.dirstribute;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/4/3
 *
 * 1.先列出所有的算子和并行度
 *
 *
 *  [source(1)]---[map1(2)]--[map2(2)]--[map3(2)]--[map4(2)]--[map5(2)]--[map6(4)]--[map7(2)]--[map8(2)]--[map9(2)]--[map10(2)]--[print(2)]
 *
 *  2.添加分发规则
 *      默认开启了operatorChain.
 *          上下游的并行度相等，分发规则就是forward,就要进行opeartorChain。
 *                  但是人为设置了禁用合并，此时不改变分发规则。
 *          上下游不相等
 *                  指定。
 *                  未指定，使用reblance。
 *
 *  [source(1)]-rb--[map1(2)]-g-[map2(2)]-hash-[map3(2)]-forward-[map4-->map5(2)]-r-[map6(4)]-r-[map7-->map8(2)]-f-[map9(2)]-f-[map10-->print(2)]
 */
public class Demo5_Exec
{
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(2);

        env.socketTextStream("hadoop102", 8888).name("source")
            .map(x -> x).name("map1").global()
            .map(x -> x).name("map2").keyBy(x -> x)
            .map(x -> x).name("map3")
            .map(x -> x).name("map4").startNewChain()
            .map(x -> x).name("map5")
            .map(x -> x).name("map6").setParallelism(6)
            .map(x -> x).name("map7")
            .map(x -> x).name("map8")
            .map(x -> x).name("map9").disableChaining()
            .map(x -> x).name("map10")
            .print();


        try {
           env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
